/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 */

package org.apache.skywalking.oap.server.core.analysis.data;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author peng-yongsheng
 */
// Window 抽象类是 MergeDataCache 缓冲区的基类，使用双缓冲队列的结构实现
public abstract class Window<DATA> {

    // 控制 pointer 指针切换的字段
    private AtomicInteger windowSwitch = new AtomicInteger(0);

    // 指向当前正在写入的缓冲队列
    // pointer 指向的队列称为 "current 队列"（一般是有空闲空间的队列，主要负责缓冲新写入数据），
    // 另一个队列称为 "last 队列"（一般填充了一定量的数据，会有其他线程从中读取数据进行消费）。
    private SWCollection<DATA> pointer;

    // A、B两个缓冲队列
    private SWCollection<DATA> windowDataA;
    private SWCollection<DATA> windowDataB;

    Window() {
        this(true);
    }

    Window(boolean autoInit) {
        if (autoInit) {
            init();
        }
    }

    protected void init() {
        this.windowDataA = collectionInstance();
        this.windowDataB = collectionInstance();
        this.pointer = windowDataA;
    }

    public abstract SWCollection<DATA> collectionInstance();

    // 检查 windowSwitch 字段，以及 last 队列是否处于可读状态
    public boolean trySwitchPointer() {
        return windowSwitch.incrementAndGet() == 1 && !getLast().isReading();
        // 如果此时 last 队列处于 reading 状态，切换后，last 队列会变成current队列，
        // 就会出现两个线程(一个读线程、一个写线程)并发操作该队列的可能，所以需要进行
        // reading 状态的检测
    }

    // 在 trySwitchPointer() 方法尝试之后，需要在 finally 代码块中恢复 windowSwitch 字段的值，为下次检查做准备
    public void trySwitchPointerFinally() {
        windowSwitch.addAndGet(-1);
    }

    public void switchPointer() {
        // 根据 pointer 当前的指向，进行修改
        if (pointer == windowDataA) {
            pointer = windowDataB;
        } else {
            pointer = windowDataA;
        }
        // 修改 last 队列的状态
        getLast().reading();
    }

    SWCollection<DATA> getCurrentAndWriting() {
        if (pointer == windowDataA) {
            windowDataA.writing();
            return windowDataA;
        } else {
            windowDataB.writing();
            return windowDataB;
        }
    }

    private SWCollection<DATA> getCurrent() {
        return pointer;
    }

    public int currentCollectionSize() {
        return getCurrent().size();
    }

    public SWCollection<DATA> getLast() {
        if (pointer == windowDataA) {
            return windowDataB;
        } else {
            return windowDataA;
        }
    }

    public void finishReadingLast() {
        getLast().clear();
        getLast().finishReading();
    }
}
